package com.icarbonx.dap.notification.job

import com.google.gson.Gson
import com.icarbonx.dap.notification.msgservice.rabbitmq.{ConnUtil, MQObject}
import com.icarbonx.dap.notification.util.{DateUtil, KafkaUtil, NoticeMsg, SparkUtil}
import com.icarbonx.dap.notification.vo._
import kafka.serializer.StringDecoder
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Duration, Seconds, StreamingContext}

import scala.collection.mutable.ArrayBuffer

/**
 * Created by Lenovo on 2018/8/12.
 * 计算运动:24小时没有运动数据,使用运动和步数两个数据源
 */
object StepsStream {
  def main(args: Array[String]) {
    Logger.getLogger("org.apache.kafka").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.zookeeper").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    println(args.mkString("\t"))
    /**
     * brokers = "10.47.250.4:9092,10.47.250.7:9092,10.47.250.6:9092"
     * topicList = "abel_test"
     * mq:10.104.103.7 5672 guest icx@betaOn16 /
     * redis:10.104.103.7 6379 passxx#$#dslllfjshwerw123@ 3000 abel_test
     */
    val Array(brokers, sugarTopicList, sportTopic, stepTopic, wristTopic, mqHost, mqPort, mqUsername, mqPassword,
    mqVirtualHost, exchangeName, mqRouteKey, checkpointDir, duration, rddDuration, firstTime, secondTime, thirdTime, appName
    ) = args
    //    val checkpointPath = "/data/streaming/checkpoint/steps_stream/"

    val conf = SparkUtil.getSparkConf(appName, Array(classOf[BloodSugarRecord], classOf[SportRecord],
      classOf[StepsRecord], classOf[WristBandRecord]))

    val sc = SparkContext.getOrCreate(conf)

    val updateFunc = (values: Seq[Long], state: Option[Long]) => {
      //保存状态
      val previousCount = state.getOrElse(0L)
      val newValue = if (values.isEmpty) 0L else values.max
      Some(math.max(newValue, previousCount))
    }

    val updateSugarFunc = (values: Seq[Long], state: Option[Long]) => {
      //保存状态
      val previousCount = state.getOrElse(0L)
      val result = if (previousCount == 0) values.min
      else if (values.isEmpty) previousCount
      else math.min(values.min, previousCount)
      Some(result)
    }

    def creatingFunc(): StreamingContext = {
      val ssc = new StreamingContext(sc, Seconds(duration.toInt))
      ssc.checkpoint(checkpointDir)
      val kafkaParams = KafkaUtil.getKafkaParam(brokers, appName)

      val mqConn = new MQObject(mqHost, mqPort.toInt, mqUsername, mqPassword, mqVirtualHost)

      val topics = sportTopic.split(",").toSet
      val stepsTopics = stepTopic.split(",").toSet
      val wristTopics = wristTopic.split(",").toSet
      val sugarTopics = sugarTopicList.split(",").toSet

      val sportMessages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams,topics )
      val stepsMessages = KafkaUtils .createDirectStream[String, String, StringDecoder, StringDecoder]( ssc, kafkaParams, stepsTopics )
      val wristMessages = KafkaUtils .createDirectStream[String, String, StringDecoder, StringDecoder]( ssc, kafkaParams, wristTopics )
      val sugarMessages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder]( ssc, kafkaParams, sugarTopics )

      val sugarBatch: DStream[(String, Long)] =
        sugarMessages
          .filter(_._2.nonEmpty)
          .flatMap { msg =>
            try {
              println("___________ msg :" + msg._2)
              val gson = new Gson()
              val voList =
                gson.fromJson(msg._2, classOf[Array[BloodSugarRecord]])
              voList.map { vo =>
                (vo.personId.toString, vo.timestamp)
              }.toList
            } catch {
              case e: Exception =>
                println("____________" + e.getMessage)
                List(("", 0L))
            }
          }
          .filter(_._1.nonEmpty)
          .updateStateByKey(updateSugarFunc)

      sugarBatch.checkpoint(Duration(rddDuration.toInt * 1000))

      val sportBatch: DStream[(String, Long)] = sportMessages.filter(_._2.nonEmpty).flatMap { msg =>
        try {
          println("___________ msg :" + msg._2)
          val gson = new Gson()
          val voList = gson.fromJson(msg._2, classOf[Array[SportRecord]])
            .filter(_.`type` == 2)
            .map { vo =>
              (vo.personId.toString, vo.endTime)
            }
          voList
        } catch {
          case e: Exception =>
            println("____________" + e.getMessage)
            List(("", 0L))
        }
      }
        .filter(_._1.nonEmpty)


      val stepBatch: DStream[(String, Long)] = stepsMessages
        .filter(_._2.nonEmpty)
        .flatMap { msg =>
          try {
            println("___________ msg :" + msg._2)
            val gson = new Gson()
            val voList = gson.fromJson(msg._2, classOf[Array[StepsRecord]])
              .map { vo =>
                (vo.personId.toString, vo.timestamp)
              }
            voList
          } catch {
            case e: Exception =>
              println("____________" + e.getMessage)
              List(("", 0L))
          }
        }
        .filter(_._1.nonEmpty)

      val wristBatch = wristMessages.filter(_._2.nonEmpty).map { msg =>
        try {
          println("___________ msg :" + msg._2)
          val gson = new Gson()
          val vo = gson.fromJson(msg._2, classOf[WristBandRecord])
          (vo.personId.toString, vo.timestamp)
        } catch {
          case e: Exception =>
            println("____________" + e.getMessage)
            ("", 0L)
        }
      }.filter(_._1.nonEmpty)

      val sportAndStep =
        sportBatch.union(stepBatch).union(wristBatch).updateStateByKey(updateFunc)

      sportAndStep.checkpoint(Duration(rddDuration.toInt * 1000))

      val sugarSport = sugarBatch.leftOuterJoin(sportAndStep)

      sugarSport.foreachRDD(mess =>
        if (!mess.isEmpty()) {
          mess.foreachPartition(lines => {
            val nowactualDate = DateUtil.getTodayDateTime()
            val mqMsg = ArrayBuffer[String]()
            val gson = new Gson()
            lines.foreach {
              case (pid, (sugarTime, sportTime)) => {
                println("____ pid :" + pid + ",,, sportTime:" + sportTime + ",,, sugarTime:" + sugarTime)
                //@todo  excel 9: 用户有启用手环，但是24小时内没有数据
                val (firstStartTime, firstEndTime) =
                  DateUtil.calStartAndDuration(firstTime, duration)
                val (secondStartTime, secondEndTime) =
                  DateUtil.calStartAndDuration(secondTime, duration)
                val (thirdStartTime, thirdEndTime) =
                  DateUtil.calStartAndDuration(thirdTime, duration)
                // 每晚8点

                if (((sportTime.isEmpty && nowactualDate - sugarTime >= 20 * 60 * 60 * 1000 && nowactualDate - sugarTime < 44 * 60 * 60 * 1000)
                  || (sportTime.isDefined && nowactualDate - sportTime.get >= 20 * 60 * 60 * 1000 && nowactualDate - sportTime.get < 44 * 60 * 60 * 1000))
                  && firstStartTime <= nowactualDate
                  && firstEndTime > nowactualDate
                ) {
                  val msg = new WarningMsg(
                    "glycemia",
                    pid.toLong,
                    NoticeMsg.miss_step._1,
                    NoticeMsg.miss_step._2,
                    NoticeMsg.miss_step._3,
                    if (sportTime.isDefined) sportTime.get.toString else sugarTime.toString,
                    NoticeMsg.miss_step._5,
                    NoticeMsg.miss_step._4,
                    NoticeMsg.miss_step._6
                  )
                  mqMsg += gson.toJson(msg)
                }



                if (((sportTime.isEmpty && nowactualDate - sugarTime >= 32 * 60 * 60 * 1000 && nowactualDate - sugarTime < 56 * 60 * 60 * 1000)
                  || (sportTime.isDefined && nowactualDate - sportTime.get >= 32 * 60 * 60 * 1000 && nowactualDate - sportTime.get < 56 * 60 * 60 * 1000))
                  && secondStartTime <= nowactualDate
                  && secondEndTime > nowactualDate
                ) {
                  val msg = new WarningMsg(
                    "glycemia",
                    pid.toLong,
                    NoticeMsg.miss_step._1,
                    NoticeMsg.miss_step._2,
                    NoticeMsg.miss_step._3,
                    if (sportTime.isDefined) sportTime.get.toString else sugarTime.toString,
                    NoticeMsg.miss_step._5,
                    NoticeMsg.miss_step._4,
                    NoticeMsg.miss_step._6
                  )
                  mqMsg += gson.toJson(msg)
                }


                if (((sportTime.isEmpty && nowactualDate - sugarTime >= 36 * 60 * 60 * 1000 && nowactualDate - sugarTime < 60 * 60 * 60 * 1000)
                  || (sportTime.isDefined && nowactualDate - sportTime.get >= 36 * 60 * 60 * 1000 && nowactualDate - sportTime.get < 60 * 60 * 60 * 1000))
                  && thirdStartTime <= nowactualDate
                  && thirdEndTime > nowactualDate
                ) {
                  val msg = new WarningMsg(
                    "glycemia",
                    pid.toLong,
                    NoticeMsg.miss_step_12._1,
                    NoticeMsg.miss_step_12._2,
                    NoticeMsg.miss_step_12._3,
                    if (sportTime.isDefined) sportTime.get.toString else sugarTime.toString,
                    NoticeMsg.miss_step_12._5,
                    NoticeMsg.miss_step_12._4,
                    NoticeMsg.miss_step_12._6
                  )
                  mqMsg += gson.toJson(msg)
                }
              }
            }
            if (mqMsg.nonEmpty) {
              val mqChannel = mqConn.connect().createChannel()
              mqMsg.foreach { msg =>
                ConnUtil.sendMsg(mqChannel, exchangeName, mqRouteKey, msg)
              }
              mqChannel.close()
            }

          })

        })

      ssc
    }



    val ssc = StreamingContext.getOrCreate(checkpointDir, creatingFunc)

    ssc.start()
    ssc.awaitTermination()
  }
}
